package sdm.sensor.temp;

import com.espertech.esper.client.*;

import java.io.File;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Scanner;

// log4j config

import org.apache.log4j.ConsoleAppender;
import org.apache.log4j.SimpleLayout;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;


/**
 * Based on http://coffeeonesugar.wordpress.com/2009/07/21/getting-started-with-esper-in-5-minutes/
 * 
 * @author wenuti
 *
 */
public class TempMain {
    
	/**
	 * Class for 'temp' events 
	 */
    public static class Temp {
    	static int count=0;
    	
        String Date="2011.01.01", Time="00:00:00";
        Double Temp;
        String ID="ID";

        public Temp(double t) {
        	Temp = t;
        	count++;
        	Logger.getRootLogger().debug("Event Object: Instanz=" + count + " Temp=" + Temp.toString());
        }
        
        public Temp(String Date, String Time, double t) {
        	this.Temp = t;
        	this.Date = Date;
        	this.Time = Time;
        	count++;
        	Logger.getRootLogger().debug("Event Object: Instanz=" + count + " Temp=" + Temp.toString());
        }
        
        public double getTemp() {return Temp;}
        public String getDate() {return Date;}
        public String getTime() {return Time;}
        public String getID() {return ID;}

        @Override
        public String toString() {
            return "Date=" + Date + " Time=" + Time + " Temp=" + Temp;
        }
    }
    
    /**
     * Send events of type temp. Construct temp object from line in csv file
     * 
     * @param cepRT
     * @param sc
     */
    public static void GenerateCSVTemp(EPRuntime cepRT, Scanner sc) {

    	String line;
    	String[] items;
    	
		line = sc.nextLine();
		items = line.split(",");
		Temp temp = new Temp(items[0], items[1], Double.parseDouble(items[2]));

		Logger.getRootLogger().info("Sending temp   > " + temp);
        cepRT.sendEvent(temp);
    }

    
    /**
     * main
     * 
     * @param args
     */
    public static void main(String[] args) {
    	    	
    	// initialize log4j
		SimpleLayout layout = new SimpleLayout();
		ConsoleAppender appender = new ConsoleAppender(layout);
		Logger.getRootLogger().addAppender(appender);
		
		// log-level DEBUG:	raw input, raw (event) output
		// log-level INFO: 	sent Input
		// log-level WARN:	formated event output
		Logger.getRootLogger().setLevel((Level) Level.INFO);


    	//The Configuration is meant only as an initialization-time object.
        Configuration cepConfig = new Configuration();
        cepConfig.addEventType("SensorTemp", Temp.class.getName());
       
        // tempCEPEngine is just some name
        EPServiceProvider cep = EPServiceProviderManager.getProvider("tempCEPEngine", cepConfig);
        EPRuntime cepRT = cep.getEPRuntime();
        EPAdministrator cepAdm = cep.getEPAdministrator();

        
        // print timestamp
		DateFormat dateFormat = new SimpleDateFormat("yyyy.MM.dd HH:mm:ss");
        Logger.getRootLogger().info(dateFormat.format(new Date()).toString());

        
        // Es sind Aggregate aus jeweils 10sec Zeitintervallen gesucht.
        // Da die Eingabe-Date fertig vorliegt, werden jeweils 20 Eingaben,
        // welche im Abstand von 0.5sec erstellt wurden, verwendet. 
        
        
        // 1.) Statement 'mean' --> get the mean of a 10sec window after every 10sec
		String meanExpression = 
			"insert into insRunAvg " +
			"select avg(temp) as runAvg " +
			"from SensorTemp.win:length_batch(20) ";
		EPStatement meanStatement = cepAdm.createEPL(meanExpression);
				
		// Listener 'mean'
		class meanCEPListener implements UpdateListener {	// Interface UpdateListener (inner class)
			public void update(EventBean[] newEvents, EventBean[] oldEvents) {
				for (EventBean event : newEvents) {
					System.out.println("(1) mean: \t" + event.get("runAvg"));
					Logger.getRootLogger().debug("(1) mean - Event received: " + event.getUnderlying());
				}
			}
		};
		
		meanStatement.addListener(new meanCEPListener());	
        
        
        // 2.) Statement 'max' --> get the max of a 10sec window after every 10sec
		String maxExpression = 
			"select max(temp) as maxTemp " +
			"from SensorTemp.win:length_batch(20) ";
		EPStatement maxStatement = cepAdm.createEPL(maxExpression);
				
		// Listener 'max'
		class maxCEPListener implements UpdateListener {
			public void update(EventBean[] newEvents, EventBean[] oldEvents) {
				for (EventBean event : newEvents) {
					System.out.println("(2) max: \t" + event.get("maxTemp"));
					Logger.getRootLogger().debug("(2) max - Event received: " + event.getUnderlying());
				}
			}
		};
		
		maxStatement.addListener(new maxCEPListener());	
        
        
		// 3.) Statement 'deltaAvg' --> get the difference of 
		//     the avg of two following 10sec windows
		String deltaExpression = 
			"select runAvg as current, prior(1,runAvg) as previous, " +
			" (runAvg-prior(1, runAvg))/prior(1,runAvg) as deltaAvg " +
			"from insRunAvg.win:length(1)";
		EPStatement deltaStatement = cepAdm.createEPL(deltaExpression);
				
		// Listener 'max'
		class deltaCEPListener implements UpdateListener {
			public void update(EventBean[] newEvents, EventBean[] oldEvents) {
				for (EventBean event : newEvents) {
					System.out.println("(3) deltaAvg: \t" + event.get("deltaAvg"));
					Logger.getRootLogger().debug("(3) delta - Event received: " + event.getUnderlying());
				}
			}
		};
		
		deltaStatement.addListener(new deltaCEPListener());	   
		
		
		
		
		
		
        // local file
		File inFile = new File("data/sdm_temperatur_recording2.csv");
		if (!inFile.exists() || !inFile.canRead() || !inFile.isFile()) {
			System.out.println("can't find input file " + inFile.getAbsolutePath());
			return;
		}
	

		try {
			Scanner sc = new Scanner(inFile, "ISO-8859-1");
			
			// Skip top three lines
			for(int i=0; i<3; i++) {
				if (sc.hasNextLine()) {
					sc.nextLine();
					Logger.getRootLogger().info("removed line " + i);
				}
			}

			// while (sc.hasNextLine())
			// 		System.out.println(sc.nextLine());
			
			// read lines till 'EOF'
			while (sc.hasNextLine()) {
				GenerateCSVTemp(cepRT, sc);
				
			}
		}
		catch (Exception e) {
			System.out.println(e.getMessage());
			e.printStackTrace();
			return;
		}	
    }
}
